/*
 * Copyright (c) 2024 Huawei Technologies Co.,Ltd.
 *
 * openGauss is licensed under Mulan PSL v2.
 * You can use this software according to the terms and conditions of the Mulan PSL v2.
 * You may obtain a copy of Mulan PSL v2 at:
 *
 * http://license.coscl.org.cn/MulanPSL2
 *
 * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
 * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
 * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
 * See the Mulan PSL v2 for more details.
 * -------------------------------------------------------------------------
 *
 * htap_xlog_redoer.cpp
 *
 * IDENTIFICATION
 * src/gausskernel/storage/htap/htap_xlog_redoer.cpp
 *
 * -------------------------------------------------------------------------
 */

#include <ostream>
#include <istream>
#include <iomanip>
#include <cstring>
#include "c.h"
#include "utils/palloc.h"
#include "htap_logger.h"
#include "htap_debug.h"
#include "htap_infra.h"
#include "postgres.h"
#include "storage/procarray.h"
#include "postmaster/postmaster.h"
#include "session_manager.h"
#include "mot_engine.h"
#include "postmaster/postmaster.h"
#include "executor/executor.h"
#include "utils/postinit.h"
#include "utils/ps_status.h"
#include "catalog/pg_database.h"
#include "catalog/pg_partition_fn.h"
#include "htap_helpers.h"
#include "access/xlogreader.h"
#include "access/xlogreader.h"
#include "access/xlog.h"
#include "access/xlogdefs.h"
#include "access/xlog_basic.h"
#include "access/imcs/imcs_ctlg.h"
#include "access/imcs/imcu_cache_mgr.h"
#include "htap_xlog_redoer.h"

typedef struct XLogPageReadPrivate {
    const char *datadir;
    TimeLineID tli;
} XLogPageReadPrivate;

using namespace std;
using namespace MOT;
namespace htap {
HTAP_DECLARE_LOGGER(HTAPXLogRedoer, HTAP)

HTAPXLogRedoer *HTAPXLogRedoer::m_xLogRedoer = nullptr;

static void XLogRedoerExit(SIGNAL_ARGS);
static void XLogRedoerGotSigHup(SIGNAL_ARGS);
static Oid GetOidFromXLogReaderState(XLogReaderState *xlogreader);

HTAPXLogRedoer *HTAPXLogRedoer::GetInstance()
{
    return m_xLogRedoer;
}

void HTAPXLogRedoer::DestroyInstance()
{
    if (m_xLogRedoer) {
        delete m_xLogRedoer;
        m_xLogRedoer = nullptr;
    }
}

RC HTAPXLogRedoer::Start(ThreadId &threadId)
{
    std::lock_guard<std::mutex> guard(m_lock);
    if (m_isRunning) {
        HTAP_REPORT_ERROR(HTAP_RC_INVALID_STATE, "HTAPXLogRedoerStart",
            "Cannot start htap xlog redoer thread: htap xlog redoer thread is already running");
        return HTAP_RC_INVALID_STATE;
    }

    // launch htap xlog redo thread as kernel thread
    RC result = HTAP_RC_OK;
    PG_TRY();
    {
        m_xLogRedoerThread = initialize_util_thread(HTAP_XLOG_REDOER);
        if (m_xLogRedoerThread == 0) {
            HTAP_REPORT_ERROR(HTAP_RC_SYSTEM_FAILURE, "HTAPXLogRedoer Start",
                "Failed to create kernel thread for htap xlog redoer");
            result = HTAP_RC_SYSTEM_FAILURE;
        } else {
            m_isRunning = true;
            threadId = m_xLogRedoerThread;
        }
    }
    PG_CATCH();
    {
        HTAP_LOG_ERROR_STACK("Failed to create htap xlog redoer thread: %s (%d)", Geterrmsg(), geterrcode());
        FlushErrorState();
    }
    PG_END_TRY();

    m_confEnableXLogRedoer = HTAPInfra::GetInstance()->GetCfg().m_enabelHTAPXLogRedoer;
    HTAP_LOG_INFO("enable htap xlog redoer conf is %d", m_confEnableXLogRedoer);

    return result;
}

RC HTAPXLogRedoer::Stop()
{
    std::lock_guard<std::mutex> guard(m_lock);
    if (!m_isRunning) {
        HTAP_REPORT_ERROR(HTAP_RC_INVALID_STATE, "HTAPXLogRedoer Stop",
            "Cannot stop htap xlog redo thread: htap xlog redo thread is not running");
        return HTAP_RC_INVALID_STATE;
    }

    // signal thread to stop
    RC result = HTAP_RC_OK;
    PG_TRY();
    {
        int res = gs_signal_send(m_xLogRedoerThread, SIGQUIT);
        if (res != 0) {
            HTAP_REPORT_SYSTEM_ERROR_CODE(res, HTAP_RC_SYSTEM_FAILURE, "HTAPXLogRedoer Stop", "gs_signal_send failed");
            result = HTAP_RC_SYSTEM_FAILURE;
        } else {
            // join thread
            gs_thread_t thr = { m_xLogRedoerThread };
            res = gs_thread_join(thr, nullptr);
            if (res != 0) {
                HTAP_REPORT_SYSTEM_ERROR_CODE(res, HTAP_RC_SYSTEM_FAILURE, "HTAPXLogRedoer Stop",
                    "gs_thread_join failed");
                result = HTAP_RC_SYSTEM_FAILURE;
            } else {
                m_isRunning = false;
                m_xLogRedoerThread = 0;
            }
        }
    }
    PG_CATCH();
    {
        HTAP_LOG_ERROR_STACK("Failed to stop htap xlog redoer thread: %s (%d)", Geterrmsg(), geterrcode());
        FlushErrorState();
    }
    PG_END_TRY();

    return result;
}

bool HTAPXLogRedoer::IsRunning()
{
    std::lock_guard<std::mutex> guard(m_lock);
    return m_isRunning;
}

ThreadId HTAPXLogRedoer::GetThreadId()
{
    std::lock_guard<std::mutex> guard(m_lock);
    return m_xLogRedoerThread;
}

void HTAPXLogRedoer::RunXLogRedoer()
{
    HTAP_LOG_INFO("HTAPXLogRedoer thread started");

    InitXLogRedoThread();
    if (t_thrd.htap_cxt.xlog_redoer_need_exit) {
        HTAP_LOG_INFO("HTAP XLog Redoer thread ordered to stop after init");
        return;
    }

    if (!HTAPInfra::HTAPTxnEnsure()) {
        return;
    }

    MOT::TxnManager *txn = HTAPGetSafeTxn();
    m_memory_controller = HTAPInfra::GetInstance()->CreateMemoryContext(
        HTAPInfra::GetInstance()->GetThreadMemoryContext(), "htapXLogRedoerCtx");
    MemoryContext saved_context = MemoryContextSwitchTo(m_memory_controller);

    // htap xlog redo main loop
    uint64_t flushTimeoutMillis = HTAPInfra::GetInstance()->GetCfg().m_htapXLogRedoTimeoutMillis;
    while (!SignaledToStop(flushTimeoutMillis)) {
        if (XLogRecPtrIsInvalid(IMCU_CACHE->m_xlog_current_lsn)) {
            continue;
        }
        if (t_thrd.htap_cxt.xlog_redoer_got_sighup) {
            t_thrd.htap_cxt.xlog_redoer_got_sighup = false;
            ProcessConfigFile(PGC_SIGHUP);
            flushTimeoutMillis = HTAPInfra::GetInstance()->GetCfg().m_htapXLogRedoTimeoutMillis;
        }

        StartTransactionCommand();
        (void)GetTransactionSnapshot();

        txn->SetSnapshot();
        // Init memory
        m_cycle_memory_controller = HTAPInfra::GetInstance()->CreateMemoryContext(
            HTAPInfra::GetInstance()->GetThreadMemoryContext(), "htapXLogRedoerCtx");
        MemoryContext saved_cycle_context = MemoryContextSwitchTo(m_cycle_memory_controller);

        if (m_confEnableXLogRedoer) {
            XLogRedo();
        }
        setCommitCsn(getLocalNextCSN());
        // restore memory_context
        MemoryContextSwitchTo(saved_cycle_context);
        HTAPInfra::GetInstance()->DestroyMemoryContext(m_cycle_memory_controller);
        CommitTransactionCommand();
    }

    // restore memory_context
    MemoryContextSwitchTo(saved_context);
    HTAPInfra::GetInstance()->DestroyMemoryContext(m_memory_controller);

    // cleanup
    HTAP_LOG_INFO("HTAPXLogRedoer thread exiting");
}

bool HTAPXLogRedoer::CreateInstance()
{
    if (m_xLogRedoer != nullptr) {
        HTAP_LOG_ERROR("HTAPXLogRedoer already created.");
        return false;
    }
    m_xLogRedoer = new (std::nothrow) HTAPXLogRedoer();

    if (m_xLogRedoer == nullptr) {
        HTAP_LOG_ERROR("HTAPXLogRedoer allocation failed.");
        return false;
    }

    return true;
}

void HTAPXLogRedoer::InitXLogRedoThread()
{
    // we are a postmaster sub-process now
    char user_name[NAMEDATALEN];
    IsUnderPostmaster = true;
    t_thrd.role = HTAP_XLOG_REDOER;

    /* reset t_thrd.proc_cxt.MyProcPid */
    t_thrd.proc_cxt.MyProcPid = gs_thread_self();

    /* record Start Time for logging */
    t_thrd.proc_cxt.MyStartTime = time(NULL);
    t_thrd.proc_cxt.MyProgName = "HTAPXLogRedoer";

    /* Identify myself via ps */
    init_ps_display("htap xlog redo in back-end process", "", "", "");

    SetProcessingMode(InitProcessing);

    /*
     * Only the signal of reloading the configuration file and exiting is processed here
     */
    (void)gspqsignal(SIGINT, SIG_IGN);
    (void)gspqsignal(SIGTERM, SIG_IGN);
    (void)gspqsignal(SIGQUIT, XLogRedoerExit);
    (void)gspqsignal(SIGALRM, SIG_IGN);

    (void)gspqsignal(SIGPIPE, SIG_IGN);
    (void)gspqsignal(SIGUSR1, SIG_IGN);
    (void)gspqsignal(SIGUSR2, SIG_IGN);
    (void)gspqsignal(SIGFPE, SIG_IGN);
    (void)gspqsignal(SIGHUP, XLogRedoerGotSigHup);
    (void)gspqsignal(SIGURG, SIG_IGN);
    /* Early initialization */
    BaseInit();

    /* Unblock signals (they were blocked when the postmaster forked us) */
    gs_signal_setmask(&t_thrd.libpq_cxt.UnBlockSig, NULL);
    (void)gs_signal_unblock_sigusr2();

    t_thrd.proc_cxt.PostInit->SetDatabaseAndUser(DEFAULT_DATABASE, InvalidOid, user_name);
    t_thrd.proc_cxt.PostInit->InitHTAPXLogRedoer();
}

bool HTAPXLogRedoer::SignaledToStop(uint64_t timeoutMillis)
{
    // wait for the given timeout for one of 2 events: latch was set, or shutdown was signaled
    bool res = false;
    int waitRes = WaitLatch(&t_thrd.proc->procLatch, WL_LATCH_SET | WL_POSTMASTER_DEATH | WL_TIMEOUT, timeoutMillis);
    if (waitRes & WL_POSTMASTER_DEATH) {
        HTAP_LOG_INFO("HTAPXLogRedoer detected postmaster death");
        res = true;
    } else if (t_thrd.htap_cxt.xlog_redoer_need_exit) {
        HTAP_LOG_INFO("HTAPXLogRedoer exit signal caught");
        res = true;
    }
    return res;
}

void XLogRedoerExit(SIGNAL_ARGS)
{
    int saveErrno = errno;

    t_thrd.htap_cxt.xlog_redoer_need_exit = true;

    if (t_thrd.proc) {
        HTAP_LOG_DEBUG("Received xlog redo exit signal");
        SetLatch(&t_thrd.proc->procLatch);
    }

    errno = saveErrno;
}

void XLogRedoerGotSigHup(SIGNAL_ARGS)
{
    int saveErrno = errno;

    t_thrd.htap_cxt.xlog_redoer_got_sighup = true;

    if (t_thrd.proc) {
        HTAP_LOG_DEBUG("Received xlog redo hangup signal");
        SetLatch(&t_thrd.proc->procLatch);
    }

    errno = saveErrno;
}

Oid GetOidFromXLogReaderState(XLogReaderState *xlogreader)
{
    uint8 info = XLogRecGetInfo(xlogreader) & ~XLR_INFO_MASK;
    uint8 action_type = info & XLOG_HEAP_OPMASK;
    if (action_type == XLOG_HEAP_INSERT || action_type == XLOG_HEAP_DELETE || action_type == XLOG_HEAP_UPDATE ||
        action_type == XLOG_HEAP_HOT_UPDATE) {
        DecodedBkpBlock *bkpb = NULL;
        if (!xlogreader->blocks[0].in_use)
            return InvalidOid;
        bkpb = &xlogreader->blocks[0];
        return bkpb->rnode.relNode;
    }
    return InvalidOid;
}

void HTAPXLogRedoer::XLogRedo()
{
    errno_t errorno = EOK;
    char *errormsg = NULL;
    XLogReaderState *xlogreader = NULL;
    XLogPageReadPrivate readprivate;
    XLogRecord *record = NULL;
    XLogRecPtr start_lsn = IMCU_CACHE->m_xlog_current_lsn;
    IMCSHashEntry *result_found = NULL;
    bool found = false;

    errorno = memset_s(&readprivate, sizeof(XLogPageReadPrivate), 0, sizeof(XLogPageReadPrivate));
    securec_check(errorno, "", "");

    readprivate.datadir = t_thrd.proc_cxt.DataDir;
    readprivate.tli = GetRecoveryTargetTLI();
    xlogreader = XLogReaderAllocate(&SimpleXLogPageRead, &readprivate);
    if (xlogreader == NULL) {
        ereport(ERROR, (errcode(ERRCODE_OUT_OF_MEMORY), errmsg("out of memory"),
            errdetail("Failed while allocating an htap XLog reading processor")));
    }

    if (0 == start_lsn % XLogSegSize) {
        XLByteAdvance(start_lsn, SizeOfXLogLongPHD);
    } else if (0 == start_lsn % XLOG_BLCKSZ) {
        XLByteAdvance(start_lsn, SizeOfXLogShortPHD);
    }

    while (XLogReadRecord(xlogreader, start_lsn, &errormsg) != NULL) {
        if (!XLogRecPtrIsInvalid(start_lsn)) {
            start_lsn = InvalidXLogRecPtr;
            continue;
        }
        IMCU_CACHE->m_xlog_current_lsn = xlogreader->ReadRecPtr;
        /* only handle heap redo xlog */
        if (XLogRecGetRmid(xlogreader) != RM_HEAP_ID) {
            continue;
        }
        Oid tableOid = GetOidFromXLogReaderState(xlogreader);
        result_found =
            (IMCSHashEntry *)hash_search(g_instance.stat_cxt.imcsHashTable, &tableOid, HASH_FIND, &found);
        if (!result_found || result_found->current_lsn == xlogreader->ReadRecPtr) {
            continue;
        }
        htap_heap_redo(xlogreader);
    }

    /* close xlogreadfd after circulation */
    CloseXlogFile();
    XLogReaderInvalReadState(xlogreader);

    if (XLogRecPtrIsInvalid(xlogreader->ReadRecPtr) && errormsg) {
        ereport(LOG, (errmsg("[htap_xlog_redoer] preparse thread get an error info %s", errormsg)));
    }
    XLogReaderFree(xlogreader);
}
} // namespace htap
